package com.elin.ch2.tools;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.function.Supplier;

public class UseCyclicBarrierNew {
    // 下标名称映射处理 流程
    // 使用ThreadLocal 保证数据线程安全
    private static ThreadLocal<Map<String, Supplier<?>>> ThreadLocalParame = ThreadLocal.withInitial(ConcurrentHashMap::new);

    public UseCyclicBarrierNew() {
    }

    public UseCyclicBarrierNew(Map<String, Supplier<?>> parame) {
        setParame(parame);
    }

    // 设置数据
    public void setParame(Map<String, Supplier<?>> parame) {
        ThreadLocalParame.set(parame);
    }

    // 核心运行方法
    public Map start() {
        try {
            return new Worker(ThreadLocalParame.get()).start();
        } finally {
            // 防止产生内存泄漏
            ThreadLocalParame.remove();
        }
    }

}

/**
 * 该对象主要使用CyclicBarrier和CountDownLatch确保在一个主线程下进行多线程协作
 * 确保在一个线程下等待多个对应子线程的数据返回操作，
 * 同时在这个线程下返回多个子线程返回的结果集合
 */
class Worker {
    //需要处理的响应数据端即返回数据时候使用的下标
    private Map<String, Supplier<?>> map;
    //栅栏
    private CyclicBarrier cb;
    // 闭锁
    private CountDownLatch cdl;
    //多个响应端返回数据的集合
    private Map result;

    public Worker(Map<String, Supplier<?>> map) {
        this.map = map;
        cb = new CyclicBarrier(map.size(), new Merge());
        cdl = new CountDownLatch(1);
    }

    // 子线程
    class SubTask extends Thread {

        private Map.Entry<String, Supplier<?>> entry;

        public SubTask(Map.Entry<String, Supplier<?>> entry) {
            this.entry = entry;
        }

        @Override
        public void run() {
            try {
//                synchronized (result) {
                //调用传递过来的key以及执行对应的逻辑操作
                    result.put(entry.getKey(), entry.getValue().get());
//                }
                // 栅栏等待
                cb.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    class Merge implements Runnable{

        @Override
        // 栅栏执行完后，需要处理闭锁的数据处理
        public void run() {
            cdl.countDown();
        }
    }

    // Worker对象核心启动方法
    public Map start() {
        result = new ConcurrentHashMap();
        // 将获取到的数据生成子线程进行处理
        for (Map.Entry<String, Supplier<?>> entry : map.entrySet()) {
            new SubTask(entry).start();
        }

        try {
            // 闭锁等待，等待完毕，各子线程的数据已经整合到result上
            cdl.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
        //返回整合数据后的result
        return result;

    }


}